package com.yanggu.spark.core.rdd.transform.keyvalue

import org.apache.spark.{SparkConf, SparkContext}

//Key-Value类型-Cogroup算子
object RDD25_Cogroup {

  def main(args: Array[String]): Unit = {

    //1. 创建sparkConf配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")

    //2. 创建spark上下文对象
    val sparkContext = new SparkContext(sparkConf)

    //3. 从内存中创建RDD
    val dataRDD1 = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("c", 3)))
    val dataRDD2 = sparkContext.makeRDD(List(("a", 1), ("c", 2), ("c", 3), ("d", 4)))

    //4. cogroup 该函数是join、leftJoin、rightJoin、FullOuterJoin的底层函数
    //函数签名: def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
    //将两个RDD中相同key对应的key分别封装到对应的迭代器中。
    //如果存在单一的key。那么有一个迭代器就没有值
    val value = dataRDD1.cogroup(dataRDD2)

    //5. 打印
    value.collect.foreach(println)

    //6. 释放资源
    sparkContext.stop()
  }

}
